package day04;

import data.ClickSource;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import po.Event;

import java.sql.Timestamp;

/**
 * @Description: 使用KeyedProcessFunction模拟5秒的滚动窗口，
 * 模拟的是增量聚合函数和全窗口聚合函数结合使用的情况
 * @Author: ZYX
 * @Date: 2022/2/11 13:49
 * @Version: 1.0
 */
public class Demo03 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new ClickSource())
                .keyBy(obj -> obj.user)
                .process(new FakeWindow())
                .print();

        env.execute();
    }

    private static class FakeWindow extends KeyedProcessFunction<String, Event,String> {
        // key是窗口的开始时间，value是窗口中的pv累加值
        private MapState<Long,Integer> mapState;
        //窗口大小
        private Long windowTimed = 5000L;


        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor<Long,Integer>("窗口时间的PVcount", Types.LONG,Types.INT));
        }

        // 累加器
        @Override
        public void processElement(Event event, Context context, Collector<String> out) throws Exception {
            // 计算当前元素所属窗口的开始时间
            long curTime = context.timerService().currentProcessingTime();
            // 计算窗口开始时间的公式
            long windowStart = curTime - curTime % windowTimed;
            long windowEnd = windowStart + windowTimed;

            if (mapState.contains(windowStart)){
                mapState.put(windowStart,mapState.get(windowStart)+1);
            }else {
                mapState.put(windowStart,1);
            }

            context.timerService().registerProcessingTimeTimer(windowEnd - 1L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            long windowEnd = timestamp + 1L;
            long windowStart = windowEnd - windowTimed;
            Integer count = mapState.get(windowStart);
            out.collect("用户：" + ctx.getCurrentKey() + " 在窗口" +
                    "" + new Timestamp(windowStart) + "~" + new Timestamp(windowEnd) + "" +
                    "中的pv次数是：" + count);
            mapState.remove(windowStart);
        }

    }
}
